package com.sheepone.distributor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sheepone.distributor.event.DistributeEvent;
import com.sheepone.distributor.event.EventContext;
import com.sheepone.distributor.thread.DelayEventQueue;
import com.sheepone.distributor.thread.NamedThreadFactoryManager;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;


/**
 * @author Miss.杨
 * @description
 * @since 2023-12-07 21:53
 */
public class DistributorLine extends DisruptorHolder<DistributeEvent> {


    private final long sleepNanos = 5;
    private final ExecutorService executorService;
    private final DelayEventQueue<EventTranslator<DistributeEvent>> delayQueue;


    public DistributorLine(String name, int size, DelayEventQueue<EventTranslator<DistributeEvent>> delayQueue) {
        super(name, DistributeEvent::new, WorkerStationManager.INSTANCE);

        this.delayQueue = delayQueue;
        executorService = Executors.newSingleThreadExecutor(NamedThreadFactoryManager.INSTANCE.acquire(name + "Delay"));

        line = newDisruptor(size);
    }

    public void publish(EventTranslator<DistributeEvent> translator) {
        //If not resizing, follow the normal publishing logic
        if (isResizing() || !line.getRingBuffer().tryPublishEvent(translator)) {
            try {
                delayQueue.put(translator);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public void stop() {
        line.halt();
        executorService.shutdown();
    }

    public void start() {
        executorService.submit(() -> {
            while (true) {
                while (!isResizing() && !delayQueue.isEmpty()) {
                    try {
                        line.getRingBuffer().publishEvent(delayQueue.take());
                    } catch (InterruptedException e) {
                        throw e;
                    }
                }
                try {
                    TimeUnit.NANOSECONDS.sleep(sleepNanos);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        line.start();
    }

}
